package com.atguigu.edu.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DwdTrafficBaseLogSplit {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // 声明消费的主题以及消费者组
        String topic = "topic_log";
        String groupId = "dwd_traffic_base_log_split_group";
        //获取消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        //kafkaStrDS.print(">>>");

        OutputTag<String> dirtyTag = new OutputTag<String>("dirtyTag"){};

        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {

                        try {
                            JSONObject jsonObj = JSON.parseObject(jsonStr);
                            out.collect(jsonObj);
                        } catch (Exception e) {
                            ctx.output(dirtyTag, jsonStr);
                        }


                    }
                }
        );

        DataStream<String> dirtyDS = jsonObjDS.getSideOutput(dirtyTag);

         dirtyDS.addSink(MyKafkaUtil.getKafkaProducer("dirtyTag"));

        KeyedStream<JSONObject, String> keyedDS = jsonObjDS.keyBy(jsonObject -> jsonObject.getJSONObject("common").getString("mid"));
        //keyedDS.print(">>>>");
       SingleOutputStreamOperator<JSONObject> fixedDS = keyedDS.map(
                new RichMapFunction<JSONObject, JSONObject>() {

                    private ValueState<String> lastVisitDateState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        lastVisitDateState = getRuntimeContext().getState(
                                new ValueStateDescriptor<String>(
                                        "lastVisitDateState",
                                        String.class
                                )
                        );
                    }

                    @Override
                    public JSONObject map(JSONObject jsonObj) throws Exception {
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        Long ts = jsonObj.getLong("ts");
                        String lastVisitDate = lastVisitDateState.value();

                        String curVisitdate = DateFormatUtil.toDate(ts);
                        if ("1".equals(isNew)) {

                            if (StringUtils.isEmpty(lastVisitDate)) {
                                lastVisitDateState.update(curVisitdate);
                            } else {
                                if (!lastVisitDate.equals(curVisitdate)) {
                                    isNew = "0";
                                    jsonObj.getJSONObject("common").put("is_new", isNew);
                                }
                            }
                        } else {
                            if (StringUtils.isEmpty(lastVisitDate)) {
                                String yesterday = DateFormatUtil.toDate(ts - 24 * 60 * 60 *1000);
                                lastVisitDateState.update(yesterday);
                            }
                        }


                        return jsonObj;
                    }
                }
        );

        OutputTag<String> errTag = new OutputTag<String>("errTag") {};
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        OutputTag<String> appVideoTag = new OutputTag<String>("appVideoTag") {};

        SingleOutputStreamOperator<String> pageDS = fixedDS.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<String> out) throws Exception {
                        JSONObject errJsonObj = jsonObj.getJSONObject("err");
                        if (errJsonObj != null) {
                            ctx.output(errTag, jsonObj.toJSONString());
                            jsonObj.remove("err");
                        }
                        JSONObject startJsonObj = jsonObj.getJSONObject("start");
                        JSONObject appVideoJson = jsonObj.getJSONObject("appVideo");
                        if (startJsonObj != null) {
                            ctx.output(startTag, jsonObj.toJSONString());

                        } else if (appVideoJson != null) {
                            ctx.output(appVideoTag, jsonObj.toJSONString());
                        } else {
                            JSONObject commonJsonObj = jsonObj.getJSONObject("common");
                            JSONObject pageJsonObj = jsonObj.getJSONObject("page");
                            Long ts = jsonObj.getLong("ts");

                            JSONArray displaysArr = jsonObj.getJSONArray("displays");
                            if (displaysArr != null && displaysArr.size() > 0) {
                                for (int i = 0; i < displaysArr.size(); i++) {
                                    JSONObject newDisplaysJsonObj = new JSONObject();
                                    JSONObject displayJsonObj = displaysArr.getJSONObject(i);
                                    newDisplaysJsonObj.put("common", commonJsonObj);
                                    newDisplaysJsonObj.put("page", pageJsonObj);
                                    newDisplaysJsonObj.put("display", displayJsonObj);
                                    newDisplaysJsonObj.put("ts", ts);
                                    ctx.output(displayTag, newDisplaysJsonObj.toJSONString());
                                }
                            }

                            JSONArray actionJson = jsonObj.getJSONArray("action");
                            if (actionJson != null && actionJson.size() > 0) {
                                for (int i = 0; i < actionJson.size(); i++) {
                                    JSONObject newActionJsonObj = new JSONObject();
                                    JSONObject displayJsonObj = displaysArr.getJSONObject(i);
                                    newActionJsonObj.put("common", commonJsonObj);
                                    newActionJsonObj.put("page", pageJsonObj);
                                    newActionJsonObj.put("action", actionJson);
                                    newActionJsonObj.put("ts", ts);
                                    ctx.output(actionTag, newActionJsonObj.toJSONString());
                                }
                            }
                            jsonObj.remove("displays");
                            jsonObj.remove("actions");
                            out.collect(jsonObj.toJSONString());


                        }
                    }
                }



        );
        DataStream<String> errDS = pageDS.getSideOutput(errTag);
        DataStream<String> startDS = pageDS.getSideOutput(startTag);
        DataStream<String> appVideoDS = pageDS.getSideOutput(appVideoTag);
        DataStream<String> displayDS = pageDS.getSideOutput(displayTag);
        DataStream<String> actionDS = pageDS.getSideOutput(actionTag);



        errDS.print("@@@@");
        startDS.print("####");
        appVideoDS.print("$$$$");
        displayDS.print("%%%%");
        actionDS.print("&&&&");
        pageDS.print(">>>>");


        pageDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_page_log"));
        errDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_err_log"));
        startDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_start_log"));
        displayDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_display_log"));
        actionDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_action_log"));
        appVideoDS.addSink(MyKafkaUtil.getKafkaProducer("dwd_traffic_appVideo_log"));

        env.execute();
    }
}
